package com.tai.mq.kafka;

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;

import java.util.Collections;
import java.util.Properties;

public class ZKConsumer {

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.56.101:9092,192.168.56.102:9092,192.168.56.103:9092");
        props.put("group.id", "myGroup_zk");
        props.put("enable.auto.commit", "false");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        TopicPartition topicPartition = new TopicPartition("myTopic1", 0);
        consumer.assign(Collections.singletonList(topicPartition));

        long lastConsumedOffset = -1;
        OffsetAndMetadata offsetAndMetadata = consumer.committed(topicPartition);
        if (offsetAndMetadata != null) {
            lastConsumedOffset = offsetAndMetadata.offset();
        }

        try {
            ZooKeeper zk = new ZooKeeper("192.168.56.101:2181,192.168.56.102:2181,192.168.56.103:2181", 3000, null);
            String path = "/consumers/myTopic_zk/offsets/myTopic1/0";
            byte[] data = Long.toString(lastConsumedOffset).getBytes();
            if (zk.exists(path, false) == null) {
                zk.create(path, data, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            } else {
                zk.setData(path, data, -1);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

